package com.shujia.sql

import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api._
import org.apache.flink.table.api.bridge.scala._

object Demo02DSToTable {
  def main(args: Array[String]): Unit = {
    val bsEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val bsSettings: EnvironmentSettings = EnvironmentSettings
      .newInstance()
      .useBlinkPlanner() // 使用Blink的SQL解析器
      .inStreamingMode() // 开启流处理模式
      .build()
    // 构建Table Env
    val bsTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(bsEnv, bsSettings)

    // 通过Socket创建一个流
    val stuDS: DataStream[(String, String, String, String, String)] = bsEnv.socketTextStream("master", 8888)
      .map(line => {
        val stuArr: Array[String] = line.split(",")
        (stuArr(0), stuArr(1), stuArr(2), stuArr(3), stuArr(4))
      })
    //    // DSL 一般在Flink中更流行SQL
    //    // 将流转化为动态表 --> Source表
    //    val stuTable: Table = stuDS.toTable(bsTableEnv, $"id", $"name", $"age", $"gender", $"clazz")
    //
    //    // 基于动态表的 连续查询 得到 结果的动态表
    //    val clazzTable: Table = stuTable
    //      .groupBy($"clazz")
    //      .select($"clazz", $"id".count())
    //
    //

    // SQL 的方式 -- Flink 主流
    bsTableEnv.createTemporaryView("stu",stuDS,$"id", $"name", $"age", $"gender", $"clazz")

    val clazzTable: Table = bsTableEnv.sqlQuery(
      """
        |select clazz,count(id) from stu group by clazz
        |""".stripMargin)


    // 将结果的动态表 转换成流
    val clazzCntDS: DataStream[(Boolean, (String, Long))] = clazzTable.toRetractStream[(String, Long)]
    clazzCntDS.print()
    bsEnv.execute()

  }


}
